package cn.azzhu.day02.transformations;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * keyBy：多字段分组
 * @author azzhu
 * @create 2020-09-18 01:00:37
 */
public class KeyByDemo3 {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //辽宁,沈阳,100
        final DataStreamSource<String> lines = env.socketTextStream("hadoop105", 9999);

//        final SingleOutputStreamOperator<Tuple3<String, String, Double>> provinceCityAndMoney = lines.map(new MapFunction<String, Tuple3<String, String, Double>>() {
//            @Override
//            public Tuple3<String, String, Double> map(String line) throws Exception {
//                final String[] fields = line.split(",");
//                final String province = fields[0];
//                final String city = fields[1];
//                final double money = Double.parseDouble(fields[2]);
//                return Tuple3.of(province, city, money);
//            }
//        });
//
//        //元组keyBy可以用角标
//        final SingleOutputStreamOperator<Tuple3<String, String, Double>> summed = provinceCityAndMoney.keyBy(0, 1).sum(2);

        final SingleOutputStreamOperator<OrderBean> provinceCityAndMoney = lines.map(new MapFunction<String, OrderBean>() {
            @Override
            public OrderBean map(String line) throws Exception {
                final String[] fields = line.split(",");
                final String province = fields[0];
                final String city = fields[1];
                final double money = Double.parseDouble(fields[2]);
                return OrderBean.of(province, city, money);
            }
        });
        final SingleOutputStreamOperator<OrderBean> summed = provinceCityAndMoney.keyBy("province", "city").sum("money");
        summed.print();

        env.execute("KeyByDemo3");
    }
}
